package org.jkuang.qstar.index.store;

import java.util.Comparator;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.jkuang.qstar.index.cmp.IndexObject;
import org.jkuang.qstar.index.iterator.NodeIterator;
import org.jkuang.qstar.index.iterator.TSIterFilter;
import org.jkuang.qstar.index.iterator.TSIterator;

/**
 * 
 * @author jkuang 2012年11月17日
 *
 * @param <T>
 */
public class DataStruct<T> extends IndexObject {

	private Comparator<T> comparator;
	private DataBlock<T>[] blocks;
	private volatile int blockssize = 0;
	private volatile int nodessize = 0;
	private final int MAX_NODES_SIZE = 0x8fff;
	private final ReentrantReadWriteLock lock;

	@SuppressWarnings("unchecked")
	public DataStruct(ReentrantReadWriteLock lock, Comparator<T> comparator, int index) {
		this.lock = lock;
		this.index = index;
		this.comparator = comparator;
		this.blocks = new DataBlock[1];
	}

	/**
	 * 统计符合条件数据有多少条
	 * 
	 * @param _s_sort_
	 * @param _e_sort_
	 * @return
	 */
	public int scanSize(T _s_sort_, T _e_sort_) {
		if (_s_sort_ == _e_sort_ && _e_sort_ == null) {
			return nodessize;
		} else if (this.comparator.compare(_s_sort_, _e_sort_) > 0) {
			return 0;
		} else {
			try {
				lock.readLock().lock();
				int _s_index = this.indexOfBlcok(_s_sort_, SerarChType._ceil);
				int _e_index = this.indexOfBlcok(_e_sort_, SerarChType._floor);
				int size = 0;
				for (int i = _s_index; i <= _e_index; i++) {
					if (i == _s_index || _s_index == _e_index) {
						size += blocks[i].size(_s_sort_, _e_sort_);
					} else {
						size += this.blocks[i].size();
					}
				}
				return size;
			} finally {
				lock.readLock().unlock();
			}
		}
	}

	/**
	 * 查找node节点属于哪个数据块
	 * 
	 * @param node
	 * @param type
	 * @return
	 */
	private int indexOfBlcok(T node, SerarChType type) {
		int fromIndex = 0;
		int toIndex = this.blocks.length - 1;
		if (node == null) {
			return type == SerarChType._ceil ? fromIndex : toIndex;
		}
		while (fromIndex <= toIndex) {
			int mid = (fromIndex + toIndex) >> 1;
			int cmp = blocks[mid].rangeContains(node);
			if (cmp < 0)
				fromIndex = mid + 1;
			else if (cmp > 0)
				toIndex = mid - 1;
			else
				return mid; // key found
		}
		switch (type) {
		case _floor:
			return toIndex;
		case _ceil:
			return fromIndex;
		default:
			if (fromIndex > this.blocks.length - 1) {
				return this.blocks.length - 1;
			} else {
				return fromIndex;
			}
		}
	}

	/**
	 * 合并数据数据块
	 * 
	 * 当索引块的数据被删空以后，需要将数据页挪出
	 * 
	 * @param index
	 * @return
	 */
	@SuppressWarnings("unchecked")
	private boolean combineBlock(int index) {
		if (blocks[index].size() == 0) {
			int numMoved = this.blockssize - index - 1;
			if (numMoved > 0) {
				DataBlock<T>[] newBlocks = new DataBlock[this.blockssize - 1];
				System.arraycopy(blocks, 0, newBlocks, 0, index);
				System.arraycopy(blocks, index + 1, newBlocks, index, numMoved);
				this.blocks = newBlocks;
				this.blockssize--;
				return true;
			} else {
				DataBlock<T>[] newBlocks = new DataBlock[this.blockssize];
				System.arraycopy(blocks, 0, newBlocks, 0, index);
				this.blocks = newBlocks;
				this.blockssize--;
				return true;
			}
		}
		return false;
	}

	/**
	 * 当页的块过大的时候，需要将页的块分裂为两个数据块
	 * 
	 * @param index
	 */
	@SuppressWarnings("unchecked")
	private void splitBlock(int index) {
		if (blocks[index].size() > MAX_NODES_SIZE) {
			DataBlock<T>[] temps = blocks[index].splitToTwo();
			DataBlock<T>[] newBlocks = new DataBlock[++this.blockssize];
			System.arraycopy(blocks, 0, newBlocks, 0, index);
			System.arraycopy(blocks, index, newBlocks, index + 1, this.blockssize - 1 - index);
			newBlocks[index] = temps[0];
			newBlocks[index + 1] = temps[1];
			this.blocks = newBlocks;
		}
	}

	/**
	 * 插入数据
	 * 
	 * @param node
	 */
	public void insert(T node) {
		try {
			lock.writeLock().lock();
			if (this.blockssize == 0) {
				this.blocks[this.blockssize++] = new DataBlock<T>(this.comparator);
				this.blocks[this.blockssize - 1].insert(node);
				this.nodessize++;
			} else if (this.blocks[this.blockssize - 1].rangeContains(node) <= 0) {
				if (this.blocks[this.blockssize - 1].insert(node)) {
					splitBlock((int) (this.blockssize - 1));
					this.nodessize++;
				}
			} else {
				int index = indexOfBlcok(node, SerarChType._index);
				if (this.blocks[index].insert(node)) {
					splitBlock(index);
					this.nodessize++;
				}
			}
		} finally {
			lock.writeLock().unlock();
		}

	}

	/**
	 * 删除数据
	 * 
	 * @param node
	 * @return
	 */
	public boolean remove(T node) {
		try {
			lock.writeLock().lock();
			if (blockssize == 0) {
				return false;
			} else if (this.blocks[blockssize - 1].rangeContains(node) == 0) {
				boolean result = this.blocks[blockssize - 1].delete(node);
				if (result) {
					combineBlock((int) (blockssize - 1));
					this.nodessize--;
				}
				return result;
			} else {
				int index = indexOfBlcok(node, SerarChType._index);
				boolean result = this.blocks[index].delete(node);
				if (result) {
					combineBlock(index);
					this.nodessize--;
				}
				return result;
			}
		} finally {
			lock.writeLock().unlock();
		}
	}

	/**
	 * 查询数据
	 * 
	 * @param node，node需要填compare中的字段
	 * @return
	 */
	public T get(T node) {
		try {
			lock.writeLock().lock();
			int low = 0;
			int high = blockssize - 1;
			while (low <= high) {
				int mid = (low + high) >> 1;
				int cmp = this.blocks[mid].rangeContains(node);
				if (cmp < 0)
					low = mid + 1;
				else if (cmp > 0)
					high = mid - 1;
				else {
					return this.blocks[mid].get(node);// (node);
				}
			}
			return null;
		} finally {
			lock.writeLock().unlock();
		}
	}

	/**
	 * 获取指定范围内的数据列表，按照升序活降序迭代输出
	 * 
	 * @param _s_sort_
	 * @param _e_sort_
	 * @param desc
	 * @param filter
	 * @return
	 */
	public TSIterator<T> blockList(T _s_sort_, T _e_sort_, boolean desc, TSIterFilter<T> filter) {
		try {
			lock.readLock().lock();
			int fromIndex = 0;
			int toIndex = this.blockssize - 1;
			if (_s_sort_ != null) {
				fromIndex = indexOfBlcok(_s_sort_, SerarChType._ceil);
			}
			if (_e_sort_ != null) {
				toIndex = indexOfBlcok(_s_sort_, SerarChType._ceil);
			}
			if (fromIndex > toIndex) {
				return null;
			} else {
				return new NodeIterator<T>(comparator, filter, blocks, fromIndex, toIndex, _s_sort_, _e_sort_, desc);
			}
		} finally {
			lock.readLock().unlock();
		}

	}

}
